package com.study.flink.java.day06_exactly;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class OperatorStateDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        env.enableCheckpointing(5000);

        env.getConfig().setRestartStrategy(org.apache.flink.api.common.restartstrategy.RestartStrategies.fixedDelayRestart(2, 2000));

        env.setStateBackend(new FsStateBackend("file:///D:\\IDEA\\flink-study\\dir\\day06\\backend"));

        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        DataStreamSource<String> lines = env.socketTextStream("node02", 8888);
        lines.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                if("error".equals(s)) {
                    throw new RuntimeException("error by hand!");
                }
                return s.toUpperCase();
            }
        }).print();

        //DataStreamSource<Tuple2<String, String>> tp = env.addSource(new MyParFileSource("dir/day06/mysource"));
        DataStreamSource<Tuple2<String, String>> tp = env.addSource(new MyParFileExactlyOnceSource("dir/day06/mysource"));

        tp.print();

        env.execute("OperatorStateDemo-java");

    }
}
